package com.swak.paxos.impl;

import java.io.Serializable;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;

import com.swak.exception.RemotingException;
import com.swak.paxos.Node;
import com.swak.paxos.PaxosConfig;
import com.swak.paxos.PaxosConfig.GroupConfig;
import com.swak.paxos.Value;
import com.swak.paxos.breakpoint.NetworkBP;
import com.swak.paxos.exception.PaxosException;
import com.swak.paxos.protocol.PaxosForward;
import com.swak.paxos.protocol.PaxosMessage;
import com.swak.paxos.store.PaxosStorage;
import com.swak.paxos.store.PaxosStorageManager;
import com.swak.reactivex.cluster.Cluster;
import com.swak.reactivex.cluster.ClusterConfig;
import com.swak.reactivex.cluster.ClusterExchangeHandler;
import com.swak.reactivex.cluster.ID;
import com.swak.reactivex.cluster.Member;
import com.swak.reactivex.transport.Channel;
import com.swak.reactivex.transport.ExchangeChannel;
import com.swak.reactivex.transport.ExchangeHandler;
import com.swak.reactivex.transport.Message;
import com.swak.utils.Ints;
import com.swak.utils.JsonMapper;
import com.swak.utils.Maps;
import com.swak.utils.StringUtils;

/**
 * PxosNode.
 * 
 * @author 618lf
 */
public class PaxosNode implements Node {

	private static final Logger logger = LoggerFactory.getLogger(Node.class);
	private static final String PaxosGroupsKey = "PaxosGroups";

	private final String nodeID;
	private final PaxosConfig config;
	private final PaxosStorageManager storageManager;
	private final AtomicBoolean started = new AtomicBoolean(false);
	private final Map<Integer, PaxosGroup> groups = Maps.newHashMap();
	private final Member member;

	public PaxosNode(PaxosConfig config) throws RemotingException {
		this(config, null);
	}

	public PaxosNode(PaxosConfig config, ExchangeHandler handler) throws RemotingException {
		Assert.notNull(config, "Config is null.");
		Assert.notNull(config.getCluster(), "Cluster Config is null.");

		// 初始化 nodeID
		String defaultNodeID = StringUtils.defaultIfBlank(config.getNodeID(), config.getCluster().getPeerID());
		this.nodeID = Node.newNodeID(defaultNodeID);
		this.config = config;
		this.storageManager = new PaxosStorageManager(config);

		// 初始化 Groups
		this.initGroups(this.config.getCluster());

		// 开启: P2P
		this.member = Cluster.join(this.config.getCluster().setPeerID(this.nodeID), new PaxosExchangeHandler(handler));

		// 设置为启动状态
		started.compareAndExchange(false, true);
	}

	private void initGroups(ClusterConfig clusterConfig) {

		// 初始化分组数据
		for (GroupConfig config : this.config.getGroups()) {

			// 创建分组
			this.groups.put(config.getGroupID(), this.createGroup(config.getGroupID(), config));
		}

		// 设置分组配置
		clusterConfig.getParameters().put(PaxosGroupsKey, JsonMapper.toJson(this.config.getGroups()));
	}

	private PaxosGroup createGroup(int groupID, GroupConfig config) {

		/*
		 * 创建 PaxosStorage
		 */
		PaxosStorage paxosStorage = this.storageManager.newPaxosStorage(groupID);

		/*
		 * 初始化 PaxosStorage
		 */
		paxosStorage.init();

		/*
		 * 创建 Group
		 */
		return new PaxosGroup(groupID, this.config, config, this, paxosStorage);
	}

	@Override
	public String id() {
		return this.nodeID;
	}

	@Override
	public Member member() {
		return this.member;
	}

	@Override
	public CompletableFuture<Void> propose(int groupID, Serializable value, int smID, boolean useMaster) {
		if (!this.groups.containsKey(groupID)) {
			throw new PaxosException("group error.");
		}
		return this.groups.get(groupID).propose(Value.of().setSmID(smID).setNodeID(this.id()).setValue(value));
	}

	@Override
	public CompletableFuture<Void> propose(int groupID, Serializable value, int smID) {
		return this.propose(groupID, value, smID, false);
	}

	@Override
	public void onMemberShipChanaged(Member member, List<Member> members, boolean add) {

		// 节点的 分组配置信息
		List<GroupConfig> groupConfigs = JsonMapper.fromJsonToList(member.id().getUrl().getParameter(PaxosGroupsKey),
				GroupConfig.class);

		// 本地分组信息更新
		for (PaxosGroup group : this.groups.values()) {

			// 分组配置
			GroupConfig config = null;
			if (groupConfigs != null && !groupConfigs.isEmpty()) {
				for (GroupConfig mconfig : groupConfigs) {
					if (group.getGroupID() == mconfig.getGroupID()) {
						config = mconfig;
						break;
					}
				}
			}

			// 节点配置
			group.onMemberShipChanaged(member, config, members, add);
		}
	}

	@Override
	public void onReceiveMessage(Message message) {

		Object data = message.getData();

		/*
		 * 收到不符合格式的消息
		 */
		if (data == null || !(data instanceof PaxosMessage) || !(data instanceof PaxosForward)) {
			return;
		}

		// 找到业务分组
		int groupID = -1;
		if (data instanceof PaxosMessage) {
			groupID = ((PaxosMessage) data).getGroupId();
		} else if (data instanceof PaxosForward) {
			groupID = ((PaxosForward) data).getGroupId();
		}

		// 没有此Group信息, 不需要处理
		if (!this.groups.containsKey(groupID)) {
			return;
		}

		// 执行业务分组
		this.groups.get(groupID).onReceiveMessage(message);
	}

	@Override
	public void broadcastMessage(PaxosMessage message) {

		NetworkBP.broadcastMessage(message);

		this.retryExecuteRunnable(() -> {
			try {
				this.member.broadcast(message);
			} catch (Exception e) {
			}
		});
	}

	@Override
	public void broadcastMessage(String nodeID, PaxosMessage message) {

		NetworkBP.broadcastMessage(message);

		Member member = this.member.members().lookup(new ID().setId(nodeID));

		if (member == null) {
			return;
		}

		this.retryExecuteRunnable(() -> {
			try {
				member.broadcast(message);
			} catch (Exception e) {
			}
		});
	}

	@Override
	public void broadcastMessageToFollower(PaxosMessage message) {

		PaxosGroup group = this.groups.get(message.getGroupId());
		if (group == null) {
			return;
		}

		Set<String> followers = group.getFollowers();
		for (String nodeID : followers) {
			this.broadcastMessage(nodeID, message);
		}
	}

	@Override
	public void sendMessage(String nodeID, PaxosMessage message) {

		NetworkBP.sendMessage(nodeID, message);

		Member member = this.member.members().lookup(new ID().setId(nodeID));

		if (member == null) {
			return;
		}

		this.retryExecuteRunnable(() -> {
			try {
				member.send(message);
			} catch (Exception e) {
			}
		});
	}

	@Override
	public void sendMessage(String nodeID, PaxosForward message) {

		Member member = this.member.members().lookup(new ID().setId(nodeID));

		if (member == null) {
			throw new PaxosException("Node :" + nodeID + " Not Exist");
		}

		this.retryExecuteRunnable(() -> {
			try {
				member.send(message);
			} catch (Exception e) {
			}
		});
	}

	private void retryExecuteRunnable(Runnable runnable) {
		this.retryExecuteRunnable(3, runnable);
	}

	private void retryExecuteRunnable(int times, Runnable runnable) {
		int tries = Ints.max(times, 3);
		while (tries > 0) {
			try {
				runnable.run();
				break;
			} catch (Exception e) {
			}
			tries--;
		}
	}

	/**
	 * Paxos 协议处理
	 * 
	 * @author 618lf
	 */
	class PaxosExchangeHandler implements ClusterExchangeHandler {

		private final ExchangeHandler delegate;

		PaxosExchangeHandler(ExchangeHandler delegate) {
			this.delegate = delegate;
		}

		@Override
		public void memberShipChanaged(Member member, List<Member> members, boolean add) {
			PaxosNode.this.onMemberShipChanaged(member, members, add);
		}

		@Override
		public void connected(Channel channel) throws RemotingException {
			if (this.delegate != null) {
				this.delegate.connected(channel);
			}
		}

		@Override
		public void disconnected(Channel channel) throws RemotingException {
			if (this.delegate != null) {
				this.delegate.disconnected(channel);
			}
		}

		@Override
		public void sent(Channel channel, Object message) throws RemotingException {
			if (this.delegate != null) {
				this.delegate.sent(channel, message);
			}
		}

		@Override
		public void received(Channel channel, Object msg) throws RemotingException {

			/*
			 * 处理 paxos 消息
			 */
			Message message = (Message) msg;
			PaxosNode.this.onReceiveMessage(message);

			/*
			 * 继续后续的处理
			 */
			if (this.delegate != null) {
				this.delegate.received(channel, message);
			}
		}

		@Override
		public CompletionStage<Object> reply(ExchangeChannel channel, Object message) throws RemotingException {
			if (this.delegate != null) {
				return this.delegate.reply(channel, message);
			}
			return CompletableFuture.completedStage(null);
		}

		@Override
		public void caught(Channel channel, Throwable t) throws RemotingException {

			if (logger.isDebugEnabled()) {
				logger.debug("Paxos Handle Error,", t);
			}

			if (this.delegate != null) {
				this.delegate.caught(channel, t);
			}
		}
	}

	@Override
	public void close() {

		if (!started.get()) {
			return;
		}

		for (;;) {

			if (!started.get()) {
				return;
			}

			if (started.compareAndExchange(true, false) == true) {
				this.confirmClosed();
				break;
			}
		}
	}

	private void confirmClosed() {
		try {
			member.close();
		} catch (Exception e) {
		}

		try {
			this.groups.values().forEach(group -> {
				group.close();
			});
		} catch (Exception e) {
		}
	}
}